package com.binance.zkconfig.processor;

import com.binance.zkconfig.app.constant.Constant;
import com.binance.zkconfig.app.property.ZkConfigPropertySource;
import com.binance.zkconfig.app.service.ZkConfig;
import com.binance.zkconfig.app.util.ZkConfigLoader;
import com.binance.zkconfig.app.util.ZkConfigUtils;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.Ordered;
import org.springframework.core.PriorityOrdered;
import org.springframework.core.env.CompositePropertySource;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.Environment;
import org.springframework.core.env.PropertySource;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Created by Fei.Huang on 2018/8/28.
 */
@Slf4j
public class ZkPropertySourcesProcessor implements BeanDefinitionRegistryPostProcessor, EnvironmentAware, PriorityOrdered {

    private static final String ZK_PROPERTY_SOURCES = "ZkPropertySources";

    private ConfigurableEnvironment environment;
    private static final List<String> ZK_CONFIG_NODES = Lists.newArrayList();

    public static List<String> getZkConfigNodes() {
        return ZK_CONFIG_NODES;
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory)
            throws BeansException {
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = (ConfigurableEnvironment) environment;
    }

    @Override
    public int getOrder() {
        return Ordered.HIGHEST_PRECEDENCE;
    }

    @Override
    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
        boolean enabled = Boolean.valueOf(environment.getProperty(Constant.ZK_CONFIG_ENABLED, Constant.TRUE_LOWER));
        if (!enabled) {
            log.info("ZkConfig disabled, due to the setting {}={}", Constant.ZK_CONFIG_ENABLED, enabled);
            return;
        }

        ZK_CONFIG_NODES.addAll(buildNodes());
        log.info("zkConfig added nodes:{}", ZK_CONFIG_NODES);

        if (!ZK_CONFIG_NODES.isEmpty()) {
            initZkPropertySources();
        }
    }

    private List<String> buildNodes() {
        String applicationName = environment.getProperty(Constant.SPRING_APPLICATION_NAME, String.class);
        String serviceApplicationNode = StringUtils.isEmpty(applicationName) ? "" : Constant.SERVICE_NODE + Constant.SLASH + applicationName;
        String[] customNodes = environment.getProperty(Constant.ZK_CONFIG_CUSTOM_NODES, String[].class);

        List<String> nodes = Lists.newArrayList();

        // Add GlobalCommonNode by default.
        nodes.add(Constant.COMMON_NODE);

        // Add serviceApplicationNode if spring.application.name exists
        if (!StringUtils.isEmpty(serviceApplicationNode)) {
            nodes.add(serviceApplicationNode);
        }

        // Add CustomNodes if needed.
        List<String> customNodeList = getCustomNodes(customNodes);
        if (!CollectionUtils.isEmpty(customNodeList)) {
            for (String customNode : customNodeList) {
                nodes.add(customNode);
            }
        }

        Collections.reverse(nodes);
        return nodes.stream().map(c -> c.startsWith(Constant.SLASH) ? c : Constant.SLASH + c).collect(Collectors.toList());
    }

    private List<String> getCustomNodes(String[] customNodes) {
        if (customNodes != null) {
            return Arrays.asList(customNodes);
        }
        return new ArrayList<>();
    }

    private void initZkPropertySources() {
        if (environment.getPropertySources().contains(ZK_PROPERTY_SOURCES)) {
            return;
        }
        long start = System.currentTimeMillis();
        CompositePropertySource composite = new CompositePropertySource(ZK_PROPERTY_SOURCES);
        String zkUrl = environment.getProperty(Constant.ZOOKEEPER_URL, String.class);
        CuratorFramework curatorFramework = ZkConfigUtils.getCuratorFramework(StringUtils.defaultString(zkUrl));
        for (String context : ZK_CONFIG_NODES) {
            Map<String, String> properties = ZkConfigLoader.loadProperties(context, curatorFramework);
            ZkConfig config = ZkConfigUtils.getConfig(context);
            config.syncProperties(properties);
            PropertySource<ZkConfig> propertySource = new ZkConfigPropertySource(context, config);
            composite.addPropertySource(propertySource);
        }
        environment.getPropertySources().addFirst(composite);
        long end = System.currentTimeMillis();
        log.info("zkConfig properties loaded, timeCost:{} ms", end - start);
    }
}
